package com.supreme.cacheserver.controller;

import com.supreme.commonboot.utils.AjaxResp;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.io.Serializable;
import java.util.List;

@RestController
@RequestMapping(value = "redis-stream")
public class RedisStreamController {

    @Resource
    private RedisTemplate<String, Serializable> redisTemplate;

    @Resource
    private List<RedisTemplate<String, Serializable>> redisTemplateList;

    /**
     * 创建消费组
     * @param key
     * @param group
     * @return
     */
    @GetMapping(value = "add/{key}/{group}")
    public AjaxResp add(@PathVariable(value = "key") String key,
                        @PathVariable(value = "group") String group) {
        String result = redisTemplate.opsForStream().createGroup(key, group);
        return new AjaxResp(result);
    }

    /**
     * 消费组信息
     * @param key
     * @param group
     * @return
     */
    @GetMapping(value = "consumers/{key}/{group}")
    public AjaxResp consumers(@PathVariable(value = "key") String key,
                        @PathVariable(value = "group") String group) {
        StreamInfo.XInfoConsumers consumers = redisTemplate.opsForStream().consumers(key, group);
        return new AjaxResp(consumers);
    }

    /**
     * 确认已消费
     * @param key
     * @param group
     * @param recordIds
     * @return
     */
    @GetMapping(value = "acknowledge/{key}/{group}/{recordIds}")
    public AjaxResp acknowledge(@PathVariable(value = "key") String key,
                                @PathVariable(value = "group") String group,
                                @PathVariable(value = "recordIds") String... recordIds) {
        Long acknowledge = redisTemplate.opsForStream().acknowledge(key, group, recordIds);
        return new AjaxResp(acknowledge);
    }

    /**
     * 删除消息，这里的删除仅仅是设置了标志位，不影响消息总长度
     * 消息存储在stream的节点下，删除时仅对消息做删除标记，当一个节点下的所有条目都被标记为删除时，销毁节点
     * @param key
     * @param recordIds
     * @return
     */
    @GetMapping(value = "delete/{key}/{recordIds}")
    public AjaxResp delete(@PathVariable(value = "key") String key,
                                @PathVariable(value = "recordIds") String... recordIds) {
        Long delete = redisTemplate.opsForStream().delete(key, recordIds);
        return new AjaxResp(delete);
    }

    /**
     * 消息长度
     * @param key
     * @return
     */
    @GetMapping(value = "size/{key}")
    public AjaxResp size(@PathVariable(value = "key") String key) {
        Long delete = redisTemplate.opsForStream().size(key);
        return new AjaxResp(delete);
    }
}
